import socket

from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession


def get_local_ip():
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("8.8.8.8", 80))
        ip = s.getsockname()[0]
        s.close()
        return ip
    except:
        return "127.0.0.1"


class RemoteSparkConfig:
    _spark_session = None


    def __init__(
            self,
            app_name: str = "RemoteSparkApp",
            spark_master: str = "spark://192.168.1.201:7077",  # ui 界面http://192.168.1.201:8081/
            driver_host: str = get_local_ip(),  # 本地IP，确保远程集群能访问
            driver_port: str = "0",  # 系统自动选择空闲端口
            executor_memory: str = "2g",
            cores_max: str = "4",
    ):
        self.app_name = app_name
        self.spark_master = spark_master
        self.driver_host = driver_host
        self.driver_port = driver_port
        self.executor_memory = executor_memory
        self.cores_max = cores_max


    def get_config(self) -> SparkSession:
        _spark_session = SparkSession.builder \
            .appName(self.app_name) \
            .master(self.spark_master) \
            .config("spark.driver.host", self.driver_host) \
            .config("spark.driver.port", self.driver_port) \
            .config("spark.executor.memory", self.executor_memory) \
            .config("spark.cores.max", self.cores_max) \
            .getOrCreate()
        return _spark_session

# # 使用配置类初始化Spark
# config = RemoteSparkConfig()
# spark = SparkSession.builder \
#     .config(conf=config.get_config()) \
#     .getOrCreate()


# # 示例：计算π值验证任务提交
# from random import random
# def calculate_pi(num_samples):
#     inside = 0
#     for _ in range(num_samples):
#         x, y = random(), random()
#         if x**2 + y**2 <= 1:
#             inside += 1
#     return 4 * inside / num_samples
#
# # 并行化任务到远程集群
# samples = 1000
# parallelize = spark.sparkContext.parallelize(range(samples), numSlices=10)
# pi = parallelize.map(lambda _: (random(), random())) \
#                 .filter(lambda x: x[0]**2 + x[1]**2 <= 1) \
#                 .count() * 4 / samples
# print(f"π ≈ {pi}")
# print("1111")
#
# if __name__ == '__main__':
#     calculate_pi(2)
